package org.arch_learn.dubbo_demo.dubbo_spi_monitor_demo.monitor_filter;

import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.common.threadpool.support.fixed.FixedThreadPool;
import org.apache.dubbo.rpc.*;
import org.joda.time.Duration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.stream.Collectors;

@Activate(group = {CommonConstants.CONSUMER})
public class TPMonitorFilter extends FixedThreadPool implements Filter, Runnable {
    private static Logger LOGGER = LoggerFactory.getLogger(TPMonitorFilter.class);

    private Map<String, Map<String, List<Long>>> map = new ConcurrentHashMap<>();

    private static Map<URL, ThreadPoolExecutor> THREAD_POOLS = new ConcurrentHashMap<>();

    public TPMonitorFilter() {
        Executors.newSingleThreadScheduledExecutor()
                .scheduleWithFixedDelay(this, 1, 5, TimeUnit.SECONDS);
    }

    //通过父类创建线程池
    @Override
    public Executor getExecutor(URL url) {
        return super.getExecutor(url);
    }


    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        String methodName = invocation.getMethodName();
        long startTime = System.currentTimeMillis();
        Result result = invoker.invoke(invocation);
        long endTime = System.currentTimeMillis();
        long duration = endTime - startTime;
        Map<String, List<Long>> statistics = map.get(methodName);
        List<Long> list;
        if (null == statistics) {
            statistics = new ConcurrentHashMap<>();
            list = new ArrayList<>();
        } else {
            list = statistics.get(String.valueOf(startTime));
            if (null == list) {
                list = new ArrayList<>();
            }
        }
        list.add(duration);
        statistics.put(String.valueOf(startTime), list);
        map.put(methodName, statistics);
        System.out.println("call method:" + methodName + " use " + duration + "ms");
        return result;
    }

    @Override
    public void run() {
        long currentTimeMillis = System.currentTimeMillis();
        map.entrySet().forEach(entry -> {
            String key = entry.getKey();
            Map<String, List<Long>> value = entry.getValue();
            List<String> keyList = value.keySet().stream()
                    .filter(k -> new Duration(Long.valueOf(k), currentTimeMillis).getMillis() <= 60000)
                    .collect(Collectors.toList());
            List<Long> list = new ArrayList<>();
            List<Long> finalList = list;
            if (keyList.isEmpty()) {
                LOGGER.info("方法{}最近一分钟内未调用", key);
            } else {
                value.entrySet().stream().filter(e -> keyList.contains(e.getKey()))
                        .forEach(en -> finalList.addAll(en.getValue()));
                list = finalList.stream().sorted().collect(Collectors.toList());

                double tp90ceil = Math.ceil(list.size() * 0.9);
                int tp90Index = new Double(tp90ceil).intValue();
                Long tp90Value = list.get(tp90Index - 1);
                double tp99ceil = Math.ceil(list.size() * 0.99);
                int tp99Index = new Double(tp99ceil).intValue();
                Long tp99Value = list.get(tp99Index - 1);
                LOGGER.info("方法{}最近一分钟内的TP90为{}，TP99为{}", key, tp90Value, tp99Value);
            }
        });
    }
}
